package com.fanshuai.spring.boot.interceptor;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.listener.RecordInterceptor;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;

@Slf4j
@Component
public class MyRecordInterceptor implements RecordInterceptor<String, String> {
    public ConsumerRecord<String, String> intercept(ConsumerRecord<String, String> record) {
        byte[] trace = record.headers().lastHeader("traceId").value();
        String traceId = new String(trace, StandardCharsets.UTF_8);
        log.info("traceId={}", traceId);

        return record;
    }
}
